package chapter5;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.junit.Test;

/**
 * @author zhangxufeng@meitunmama.com 2019-08-31 15:04:41
 */
public class ConsumerAssignApp extends AbstractConsumer {

  @Test
  public void testAssign() {
    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
    List<TopicPartition> topicPartitions = new ArrayList<>();
    List<PartitionInfo> partitions = consumer.partitionsFor("test-topic");
    if (null != partitions) {
      partitions.forEach(partition -> {
        topicPartitions.add(new TopicPartition(partition.topic(), partition.partition()));
      });

      consumer.assign(topicPartitions);

      while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
        records.forEach(this::printRecord);
      }
    }
  }
}
